/**
 * Copyright [2010] Delta Air Lines, Inc. All Rights Reserved Access,
 * Modification, or Use Prohibited Without Express Permission of Delta Air
 * Lines.
 */
package com.delta.rafg.core.database.impl;


import java.io.BufferedReader;
import java.io.BufferedWriter;
import java.io.File;
import java.io.FileNotFoundException;
import java.io.FileReader;
import java.io.FileWriter;
import java.io.IOException;
import java.sql.Connection;
import java.sql.DatabaseMetaData;
import java.sql.ResultSet;
import java.sql.ResultSetMetaData;
import java.sql.SQLException;
import java.sql.Statement;
import java.util.ArrayList;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;

import com.delta.rafg.core.database.DataHandlerOptions;
import com.delta.rafg.core.database.IDataHandler;
import com.delta.rafg.core.database.impl.util.DatabaseUtils;
import com.delta.rafg.core.exception.RAFGException;
import com.delta.rafg.core.log.RAFGLogger;


/**
 * @author RAFG Development Team (D9127520)
 */
public class JavaDBSeedDataHandlerImpl extends ADataHandler {

    /**
     * Holds the value of serialVersionUID property.
     */
    private static final long serialVersionUID = 8169103414462620291L;

    /**
     * Holds the value of derbySqlConnection property.
     */
    private java.sql.Connection derbySqlConnection = null;

    /**
     * Holds the value of seedFile property.
     */
    private String seedFile = "rafg_ref_seed.sql";

    private static final String DEFAULT_SEED_FILE = "rafg_ref_seed.sql";
    private static final String DEFAULT_DEST_PATH = "rafg_seed_data";

    /**
     * Holds the Path to the seed data.
     */
    private String seedPath;

    public enum SEED_DATA_OPT {
        DEST_PASS, SOURCE_URL, SOURCE_PROPID, SOURCE_USER, DEST_PATH,
        SOURCE_PASS, SEED_FILE, DEST_USER, DEST_SCHEMA_SUFFIX,
        DEST_SERVICE, DEST_DB_TYPE;
    }

    /**
     * Holds the value of bDebugLog property.
     */
    //private final boolean bDebugLog = true;

    /**
     * This method close the connection.
     */
    @Override
    public void closeConnection() {
        throw new java.lang.UnsupportedOperationException(
            "closeConnection() not implemented.");
    }

    /**
     * This method used to get the connection.
     * @return Connection
     * @throws RAFGException RAFGException
     */
    @Override
	public final Connection getConnection() throws RAFGException {

        throw new java.lang.UnsupportedOperationException(
            "getConnection() not implemented.");
    }

    /**
     * Initialize the seed data handler. Populate the derby database connection
     * specified in the DataHandlerOptions with the seed data. The seed file is
     * read and executed against the cache table and must include DROP TABLE,
     * CREATE TABLE, INSERT INTO commands for each table. Any
     * "table does not exist" errors on the drop table commands will be ignored.
     * @param options - DataHandlerOptions for initializing SeedDataHandler
     *            <p>
     *            <b>Required DataHandlerSpecificOptions</b> <br>
     *            "CacheSqlConnection" - java.sql.Connection handle to Derby
     *            Cache SQL Connection.
     *            <p>
     *            <b>Optional DataHandlerOptions custom mappings</b> <br>
     *            "SeedFile" - Override default seed file name and location.
     * @throws RAFGException RAFGException
     */
    @Override
	public final void initialize(final DataHandlerOptions options)
        throws RAFGException {

        super.initialize(options);

        if (getDataHandlerOptions().hasDataHandlerSpecificOptionsMap()) {
            derbySqlConnection =
                (java.sql.Connection) getDataHandlerOptions()
                    .getDataHandlerSpecificOptionsMap().get(
                        "CacheSqlConnection");
        }
        if (derbySqlConnection == null) {
        	throw new RAFGException("DataHandlerOptions" + ".DataHandlerSpecificOptionsMap "
                + "must contain a \"CacheSqlConnection\"");
        }

        final String overideSeedFile =
            (String) getDataHandlerOptions().getDataHandlerSpecificOptionsMap()
                .get("SeedFile");
        if (overideSeedFile != null) {
            seedFile = overideSeedFile;
        }

        seedPath = (String) getDataHandlerOptions().getDataHandlerSpecificOptionsMap().get("SeedPath");

        if (seedPath == null) {
            throw new RAFGException("seedPath MUST be specified so I know where to read the cache seed data from.");
        }

        loadSeedData();
    }

    /**
     * The database seed file contains sql commands that can be inserted into
     * the derby database. It also contains the CREATE table command in the seed
     * file.
     * @throws RAFGException RAFGException
     */
    private void loadSeedData() throws RAFGException {

        // the database seed file contains sql commands that
        // can be inserted into the derby database.
        // it also contains the CREATE table command in the sql.seed file...
        try {
        	File sqlFile = new File(seedPath + "/" + seedFile);
        	processSeedFile(derbySqlConnection, sqlFile);

        } catch (final FileNotFoundException e) {
            throw new RAFGException("File not found for creating cache: "
                + e.getMessage());
        } catch (final IOException e) {
            e.printStackTrace();
        } catch (final Exception e) {
            e.printStackTrace();
        }
    }

	private void processSeedFile(Connection derbySqlConnection, File sqlFile) throws IOException, SQLException {
        String inputLine;
    	final BufferedReader bufReader = new BufferedReader(new FileReader(sqlFile));
        final java.sql.Statement stmt = derbySqlConnection.createStatement();
        while ((inputLine = bufReader.readLine()) != null) {
            if (inputLine != null && inputLine.length() > 1) {
            	//if the line starts with a @ its a link to another sql file, open that file and process it.
            	if (inputLine.startsWith("@")) {
            		String linkedFile = inputLine.substring(1);
//            		RAFGLogger.info("Processing " + linkedFile);
            		processSeedFile(derbySqlConnection, new File(seedPath + "/" + linkedFile));
            	} else {
	                // if the statement is a bulk import statement,
	                // change the ${destPath} reference
	                // to the same path that I am using for the SQL command
	                // file.
	                if (inputLine.contains("SYSCS_UTIL.SYSCS_IMPORT_TABLE")) {
	                    inputLine = inputLine.replace("${destPath}", seedPath);
	                }
	                try {
	                    stmt.executeUpdate(inputLine);
	                } catch (final SQLException e) {
	                    // check for table not found exception
	                    // on DROP TABLE command.
	                    // ignore if it is detected.
	                    if (inputLine.contains("DROP TABLE ")
	                        && e.getSQLState().equals("42Y55")) {
	                        // do nothing, allow through
	                    } else if (inputLine.contains("CREATE SCHEMA ")
	                        && e.getSQLState().equals("X0Y68")) {
	                        // do nothing, allow through
	                    } else {
	                        throw new SQLException(e);
	                    }
	                }
            	}
            }
        }
        stmt.close();
        bufReader.close();
	}
	
    private String getDropTableString(String fullyQualifiedTableName) {
        StringBuffer sqlCmdSb = new StringBuffer();
        sqlCmdSb.append("DROP TABLE ");
        sqlCmdSb.append(fullyQualifiedTableName);
        return sqlCmdSb.toString();
    }
    
    private String getCreateTableString(Set<String> primaryKeys, ResultSet queryResultSet, String fullyQualifiedTableName) throws SQLException {

    	int nbrNewlines = 0;

        StringBuffer sqlCmdSb = new StringBuffer();
        sqlCmdSb.append("CREATE TABLE ");
        sqlCmdSb.append(fullyQualifiedTableName);
        sqlCmdSb.append(" (");

        ResultSetMetaData meta = queryResultSet.getMetaData();
        int cols = meta.getColumnCount();
        for (int i = 1; i <= cols; i++) {
            String name = meta.getColumnName(i);
            sqlCmdSb.append(name + " ");
            String type = meta.getColumnTypeName(i);
            int size = meta.getColumnDisplaySize(i);
            int scale = meta.getScale(i);
            int nullable = meta.isNullable(i);

            if (type.equalsIgnoreCase("VARCHAR2")) {
                type = "VARCHAR";
            } else if (type.equalsIgnoreCase("NUMBER")) {
                type = "NUMERIC";
            } else if (type.equalsIgnoreCase("LONG")) {
                type = "CLOB";
            }

            sqlCmdSb.append(type);
            if (size > 0 && !type.contains("LOB") && !type.contains("DATE")) {
                if (scale <= 0) {
                    sqlCmdSb.append("(" + size + ")");
                } else {
                    sqlCmdSb.append("(" + size + ", " + scale + ")");
                }

            }

            if (nullable == ResultSetMetaData.columnNoNulls) {
                sqlCmdSb.append(" NOT NULL");
            }

            if (i < cols) {
                sqlCmdSb.append(",");
            }

            //check if I need to add a newline - only if the line has grown more than 512 bytes.  There is a line limit...
            //since ";" terminates the command, you can split lines up...
            long currNbrNewLinesNeeded = sqlCmdSb.length() / 512;
            if (currNbrNewLinesNeeded > nbrNewlines) {
                nbrNewlines++;
                sqlCmdSb.append("\n");
            }
        }

        boolean bFirst = true;
        if (primaryKeys.size() > 0) {
            sqlCmdSb.append(", PRIMARY KEY (");
            for (String key : primaryKeys) {
                if (!bFirst) {
                    sqlCmdSb.append(", ");
                } else {
                    bFirst = false;
                }
                sqlCmdSb.append(key);
            }
            sqlCmdSb.append(")");
        }

        sqlCmdSb.append(")");

        return sqlCmdSb.toString();
    }
    
    private void exportSchemaInfo(List<String> schemasToCreate, BufferedWriter bw, String destSchemaSuffix) throws IOException {
        for (String schema : schemasToCreate) {
            bw.write("CREATE SCHEMA " + schema + destSchemaSuffix);
            bw.newLine();
        }
        bw.newLine();
    }
    
    /**
     * @param srcConnection SqlConnection object of source database
     * @param fullyQualifiedTableName  table to pull data from
     * @throws SQLException  thrown due to any database errors
     * @throws IOException  thrown due to any output file write errors
     * @throws RAFGException
     */
    private void exportTableData(BufferedWriter sqlWriter, Connection srcConnection,
            String fullyQualifiedSourceTableName, String destSchemaSuffix, String tableSeedFileName) throws SQLException, IOException, RAFGException {

        String schemaName = DatabaseUtils.getSchemaName(fullyQualifiedSourceTableName);
        String tableName = DatabaseUtils.getTableName(fullyQualifiedSourceTableName);

        String fullyQualifiedTableName = schemaName + destSchemaSuffix + "." + tableName;

//        if (bTestMode) {
//            System.out.println("Exporting " + fullyQualifiedTableName + " with TestMode data altering enabled.");
//        } else {
              RAFGLogger.info("Exporting " + fullyQualifiedTableName + ".");
//        }

        FileWriter fw = new FileWriter(tableSeedFileName);
        BufferedWriter dataSeedWriter = new BufferedWriter(fw);

        //determine primary keys
        DatabaseMetaData dbMeta = srcConnection.getMetaData();
        Set<String> primaryKeys = new HashSet<String>();
        ResultSet dbMetaRs = dbMeta.getPrimaryKeys(null, schemaName, tableName);
        while (dbMetaRs.next()) {
            String columnName = dbMetaRs.getString("COLUMN_NAME");
            primaryKeys.add(columnName);
        }

        //determine if any of the columns are LOBs
        boolean bHasLob = false;
        dbMetaRs = dbMeta.getColumns(null, schemaName, tableName, null);
        while (dbMetaRs.next()) {
            String columnType = dbMetaRs.getString("TYPE_NAME");
            if (columnType.contains("LOB")) {
                bHasLob = true;
            }
        }

        StringBuffer sqlCmd = new StringBuffer();
        if (primaryKeys.size() > 0) {
            sqlCmd.append("SELECT ");
            if (!bHasLob) {
                sqlCmd.append("DISTINCT ");
            }
            sqlCmd.append("* FROM " + fullyQualifiedSourceTableName + " ORDER BY ");
            boolean bFirst = true;
            for (String k : primaryKeys) {
                if (!bFirst) {
                    sqlCmd.append(", ");
                }
                sqlCmd.append(k);
                bFirst = false;
            }

        } else {
            sqlCmd.append("SELECT * FROM " + fullyQualifiedSourceTableName);
        }
        Statement sm = srcConnection.createStatement();
        sm.execute(sqlCmd.toString());
        ResultSet rs = sm.getResultSet();

        sqlWriter.write(getDropTableString(fullyQualifiedTableName));
        sqlWriter.newLine();
        String createTblCmd = getCreateTableString(primaryKeys, rs, fullyQualifiedTableName);
        sqlWriter.write(createTblCmd);
        sqlWriter.newLine();

        //Create write the header
        ResultSetMetaData meta = rs.getMetaData();
        int cols = meta.getColumnCount();

//        boolean bAlterData = false;  //used to determine if data should be altered when test mode is on.

        //write the rows of data
        while (rs.next()) {
/*        	
            if (bTestMode) {
                bAlterData = testModeAlterDataCalc(5);  //5% chance to not insert this line.
            }
            if (!bAlterData) {
*/            
/*
                seedWriter.append("INSERT INTO ");
                seedWriter.append(fullyQualifiedTableName);
                seedWriter.append(" (");
                for (int i = 1; i <= cols; i++) {
                    seedWriter.write(meta.getColumnName(i));
                    if (i != cols) {
                        seedWriter.write(',');
                    }
                }
                seedWriter.append(") VALUES (");
*/
                for (int i = 1; i <= cols; i++) {
                    //use the destination database type since I am writing seed data for it
//                    boolean bApplyTestMode = false;
//                    if (bTestMode) {
//                        bApplyTestMode = !primaryKeys.contains(meta.getColumnName(i)) && testModeAlterDataCalc(5);
//                    }
                    String dataStr = DatabaseUtils.translateColumnData(rs.getObject(i), "DERBY", false, meta.getColumnDisplaySize(i), false);
                    dataSeedWriter.write(dataStr);
                    if (i != cols) {
                        dataSeedWriter.write(',');
                    }
                }
//                seedWriter.append(")");
                dataSeedWriter.newLine();
//            }
        }
        rs.close();
        sm.close();

        dataSeedWriter.close();
        fw.close();
    }
    
    public void extractSeedData(IDataHandler sourceDataHandler, Map<SEED_DATA_OPT, String> seedOpts, boolean bAppend) {
        try {
            List<String> createdSchemas = new ArrayList<String>();

            Connection srcConnection = sourceDataHandler.getConnection();

            String destPath = seedOpts.get(SEED_DATA_OPT.DEST_PATH);
            String seedFile = seedOpts.get(SEED_DATA_OPT.SEED_FILE);
            String destSchemaSuffix = seedOpts.get(SEED_DATA_OPT.DEST_SCHEMA_SUFFIX);

            if (destPath == null) {
                destPath = DEFAULT_DEST_PATH;
            }
            if (seedFile == null) {
                seedFile = DEFAULT_SEED_FILE;
            }
            //if this is set to create a new file, clear the createdSchemas member too
            if (!bAppend) {
            	createdSchemas.clear();
            }
            
            File outputSeedFile = new File(destPath + "/" + seedFile);
            outputSeedFile.getParentFile().mkdirs();
            FileWriter fw = new FileWriter(outputSeedFile, bAppend);
            BufferedWriter sqlWriter = new BufferedWriter(fw);

            List<String> schemasToCreate = new ArrayList<String>();
            for (String t : sourceDataHandler.getMonitoredTables()) {
                String s = DatabaseUtils.getSchemaName(t);
                if (!schemasToCreate.contains(s) && !createdSchemas.contains(s)) {
                    schemasToCreate.add(s);
                    createdSchemas.add(s);
                }
            }
            if (schemasToCreate.size() > 0) {
                exportSchemaInfo(schemasToCreate, sqlWriter, destSchemaSuffix);
            }

            for (String fullyQualifiedTableName : sourceDataHandler.getMonitoredTables()) {

                String tableName = DatabaseUtils.getTableName(fullyQualifiedTableName);
                String schemaName = DatabaseUtils.getSchemaName(fullyQualifiedTableName);
                String destFullyQualifiedTableName = schemaName + destSchemaSuffix + "." + tableName;

                String seedDataFileName = sourceDataHandler.getDataHandlerOptions().getDataHandlerPropertyIdentifier()
                        + "_" + destFullyQualifiedTableName + ".dat";

                String seedDataSqlFileName = sourceDataHandler.getDataHandlerOptions().getDataHandlerPropertyIdentifier()
                		+ "_" + destFullyQualifiedTableName + ".sql";

                FileWriter sqlFw = new FileWriter(new File(destPath + "/" + seedDataSqlFileName));
                BufferedWriter sqlBw = new BufferedWriter(sqlFw);

                //create comma delimited data file for this table
                exportTableData(sqlBw, srcConnection, fullyQualifiedTableName, destSchemaSuffix, destPath + "/" + seedDataFileName);

                //add import command to master sql file
                sqlBw.append("CALL SYSCS_UTIL.SYSCS_IMPORT_TABLE ('"
                        + schemaName + destSchemaSuffix + "', '"
                        + tableName + "', '"
                        + "${destPath}/" + seedDataFileName
                        + "', NULL, NULL, "
                        + "'utf8', 1)");
                sqlBw.newLine();
                sqlBw.close();
                sqlFw.close();
                sqlWriter.append("@" + seedDataSqlFileName);
                sqlWriter.newLine();
            }

            sqlWriter.close();
            fw.close();

        } catch (java.sql.SQLException e) {
            e.printStackTrace();
        } catch (IOException e) {
            e.printStackTrace();
        } catch (RAFGException e) {
            e.printStackTrace();
        }
    }
    
    //These don't apply here
	@Override
	public POLL_STYLE getPollStyle() {
		return null;
	}

	@Override
	public String getPollFrequency() {
		return null;
	}
}